/*******************************************************************************
 * Copyright (c) 2012, 2017 IBM Corp.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 *
 * The Eclipse Public License is available at
 *   http://www.eclipse.org/legal/epl-v10.html
 * and the Eclipse Distribution License is available at
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Ian Craggs - initial contribution
 *******************************************************************************/


#include "yuan1.h"


//接收内存
void *recv_neicun_thread(void *arg)
{
    int id = *((int *)arg);
    free(arg);  // 参数由主线程动态分配，线程内部释放

    if (id < 0 || id > 1) {
        printf("接收者编号只能是 0 或 1\n");
        pthread_exit(NULL);
    }

    int shmid = shmget(SHM_KEY, sizeof(struct shmseg), 0666);
    if (shmid == -1) {
        perror("shmget");
        pthread_exit(NULL);
    }

    int semid = semget(SEM_KEY, 1, 0666);
    if (semid == -1) {
        perror("semget");
        pthread_exit(NULL);
    }

    struct shmseg *shm = (struct shmseg *)shmat(shmid, NULL, 0);
    if (shm == (void *)-1) {
        perror("shmat");
        pthread_exit(NULL);
    }

     while (running) {
        char local_buf[SHM_SIZE] = {0};

        sem_wait(semid);

        if (shm->written == 1 && shm->read_flags[id] == 0) {
            //strncpy(local_buf, shm->text, SHM_SIZE);
            strncpy(local_buf, shm->text, SHM_SIZE - 1);
            local_buf[SHM_SIZE - 1] = '\0';

            shm->read_flags[id] = 1;

            if (shm->read_flags[0] == 1 && shm->read_flags[1] == 1) {
                shm->written = 0;
            }
        }

        sem_signal(semid);

        if (local_buf[0] != '\0') {
            printf("接收者%d 收到消息: %s\n", id, local_buf);

            // 解析 JSON
            cJSON *root = cJSON_Parse(local_buf);
            if (!root) {
                printf("JSON 解析失败！\n");
                continue;
            }

            // 提取字段
            cJSON *TEM = cJSON_GetObjectItem(root, "TEM");
            cJSON *HUM = cJSON_GetObjectItem(root, "HUM");
            cJSON *FAN = cJSON_GetObjectItem(root, "FAN");
            cJSON *LED = cJSON_GetObjectItem(root, "LED");
            cJSON *VOICE = cJSON_GetObjectItem(root, "VOICE");
            // 创建新 JSON
                cJSON *root1 = cJSON_CreateObject();
            if(FAN)
            {
                if(strcmp(FAN->valuestring,"on")==0)
                {
                    cJSON_AddBoolToObject(root1,"fan",cJSON_True);
                }
                else{
                    cJSON_AddBoolToObject(root1,"fan",cJSON_False);
                }
            }
             if(LED)
            {
                if(strcmp(LED->valuestring,"on")==0)
                {
                    cJSON_AddBoolToObject(root1,"lamp",cJSON_True);
                }
                else{
                    cJSON_AddBoolToObject(root1,"lamp",cJSON_False);
                }
            }
              if(VOICE)
            {
                if(strcmp(VOICE->valuestring,"on")==0)
                {
                    cJSON_AddBoolToObject(root1,"alarm",cJSON_True);
                }
                else{
                    cJSON_AddBoolToObject(root1,"alarm",cJSON_False);
                }
            }
            if (TEM) {
                printf("温度: %d\n", TEM->valueint);
               // printf("SHI度: %d\n", HUM->valueint);

                
                cJSON_AddNumberToObject(root1, "tem", TEM->valueint);
               // cJSON_AddNumberToObject(root1, "hum", HUM->valueint);
             
            }
		if (HUM) {
                //printf("温度: %d\n", TEM->valueint);
                printf("SHI度: %d\n", HUM->valueint);

                
                //cJSON_AddNumberToObject(root1, "tem", TEM->valueint);
               cJSON_AddNumberToObject(root1, "hum", HUM->valueint);
             
            }
		 char *new_json = cJSON_Print(root1);
                printf("新 JSON: %s\n", new_json);

                // 可以发送 new_json 到共享内存或网络
                 pthread_mutex_lock(&global_send_mutex);
            strncpy(global_send_buffer, new_json, SHM_SIZE-1);
            global_send_buffer[SHM_SIZE - 1] = '\0';
            pthread_mutex_unlock(&global_send_mutex);
                //strncpy(global_send_buffer,new_json,512);
                free(new_json);
                cJSON_Delete(root1);

            cJSON_Delete(root);  // 释放解析的 JSON
        }

        usleep(100000);
    }

    shmdt(shm);
    pthread_exit(NULL);
}

//发送队列信息函数
void *send_duilie_thread(void *arg)
{
    int msgid = msgget(MSG_KEY, 0666 | IPC_CREAT);
    if (msgid == -1) {
        perror("msgget failed");
        pthread_exit(NULL);
    }

    struct msgbuf msg;
    msg.mtype = 1;

    while (running)
    {
        pthread_mutex_lock(&global_recv_mutex);

        if (strlen(global_recv_buffer) > 0) {
            strncpy(msg.mtext, global_recv_buffer, sizeof(msg.mtext) - 1);
            global_recv_buffer[0] = '\0';  // 清空，防止重复发送

            pthread_mutex_unlock(&global_recv_mutex);

            if (msgsnd(msgid, &msg, strlen(msg.mtext) + 1, 0) == -1) {
                perror("msgsnd failed");
                break;
            }

            printf("已通过消息队列发送: %s\n", msg.mtext);
        } else {
            pthread_mutex_unlock(&global_recv_mutex);
            usleep(100 * 1000);  // 100ms 避免忙等
        }
    }

    pthread_exit(NULL);
}



void mq_send()
{
     if (strlen(global_send_buffer) > 0) {
        MQTTClient_message pubmsg = MQTTClient_message_initializer;
        pubmsg.payload = global_send_buffer;
        pubmsg.payloadlen = (int)strlen(global_send_buffer);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;

        MQTTClient_deliveryToken token;
        int rc = MQTTClient_publishMessage(client, TOPIC_send, &pubmsg, &token);
        if (rc != MQTTCLIENT_SUCCESS) {
            printf("Publish failed: %d\n", rc);
        } else {
            // 清空 buffer，防止重复发
            global_send_buffer[0] = '\0';
        }
    }
}

// 发送线程函数mqtt
void *send_thread(void *arg)
{
   while (running) {
    pthread_mutex_lock(&global_send_mutex);
    pthread_mutex_lock(&client_mutex);
    
    mq_send();
      
    
    pthread_mutex_unlock(&global_send_mutex);
    pthread_mutex_unlock(&client_mutex);
    sleep(1);  // 控制发送频率
}


   
}

int main(int argc, char *argv[])
{
    int rc = 0;
    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    // 订阅线程//订阅的是mqtt主题
    pthread_mutex_lock(&client_mutex);
    MQTTClient_subscribe(client, TOPIC_recv, QOS);
    pthread_mutex_unlock(&client_mutex);
    //


    //pthread_t recv_neicun;
int *receiver_id = malloc(sizeof(int));//内存线程传入参数
*receiver_id = 1;  // 或 1
//pthread_create(&recv_neicun, NULL, recv_neicun_thread, receiver_id);

    //

    // 创建线程
    pthread_t  send_tid,send_duilie,recv_neicun;
    //pthread_create(&recv_tid, NULL, receive_thread, NULL);
    pthread_create(&send_tid, NULL, send_thread, NULL);
    pthread_create(&send_duilie, NULL, send_duilie_thread, NULL);
    pthread_create(&recv_neicun, NULL, recv_neicun_thread, receiver_id);
    // 主线程控制退出
    // 不再 getchar()，由输入 exit 控制退出
while (running) {
    sleep(1);
}
  
    
    
    pthread_join(send_tid, NULL);
    //pthread_join(recv_tid, NULL);
    pthread_join(send_duilie,NULL);
    pthread_join(recv_neicun,NULL);
    MQTTClient_unsubscribe(client, TOPIC_recv);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    pthread_mutex_destroy(&client_mutex);
    pthread_mutex_destroy(&global_send_mutex);
    pthread_mutex_destroy(&global_recv_mutex);
    return 0;
}
